#pragma once
#include "../mqcomm/mq_log.hpp"
#include "../mqcomm/mq_helper.hpp"
#include "../mqcomm/mq_msg.pb.h"

#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
#include <cassert>
#include <algorithm>

namespace ns_kkymq
{
    struct MsgQueue
    {
        using ptr = std::shared_ptr<MsgQueue>;
        // 队列名字
        std::string _name;
        // 队列持久化标志
        bool _durable;
        // 队列独占标志
        bool _exclusive;
        // 是否自动删除标志
        bool _auto_delete;
        google::protobuf::Map<std::string, std::string> _args;

        MsgQueue() = default;

        MsgQueue(const std::string &name, bool durable, bool exclusive, bool auto_delete,
                 const google::protobuf::Map<std::string, std::string> &args)
            : _name(name), _durable(durable), _exclusive(exclusive), _auto_delete(auto_delete), _args(args)
        {
        }

        // args存储键值对，在存储数据库的时候，会组织一个格式字符串进行存储 key1=val1&key2=val2
        // 内部解析str_args字符串，将内容存储到成员中
        void setArgs(const std::string &str_args)
        {
            // key1=val1&key2=val2
            std::vector<std::string> sub_args;
            StrHelper::split(str_args, "&", sub_args);
            for (auto &str : sub_args)
            {
                size_t pos = str.find('=');
                std::string key = str.substr(0, pos);
                std::string value = str.substr(pos + 1);
                _args.insert({key, value});
            }
        }

        // 将args中的内容进行序列化后，返回一个字符串
        std::string getArgs()
        {
            if (_args.empty())
                return "";
            std::string result;
            for (auto start = _args.begin(); start != _args.end(); start++)
            {
                result += start->first + '=' + start->second + "&";
            }
            result.pop_back();
            return result;
        }

        ~MsgQueue()
        {
        }
    };
    using QueueMap = std::unordered_map<std::string, MsgQueue::ptr>;
    class MsgQueueMapper
    {
    public:
        MsgQueueMapper(const std::string dbfile)
            : _sql_helper(dbfile)
        {
            std::string path = FileHelper::parentDirectory(dbfile);
            FileHelper::createDirectory(path);
            _sql_helper.open();
            createTable();
        }

        void createTable()
        {
            //"create table if not exists queue_table(name varchar(32) primary key,
            // durable,exclusive,auto_delete,args)";
            std::stringstream sql;
            sql << "create table if not exists queue_table(";
            sql << "name varchar(32) primary key,";
            sql << "durable int,";
            sql << "exclusive int,";
            sql << "auto_delete int,";
            sql << "args varchar(128));";
            assert(_sql_helper.exec(sql.str(), nullptr, nullptr));
        }

        void removeTable()
        {
            std::string sql = "drop table if exists queue_table";
            _sql_helper.exec(sql, nullptr, nullptr);
        }

        bool insert(MsgQueue::ptr &queue_ptr)
        {
            // insert into queue_table values('queue1',true,false,false,"k1=v1&k2=v2")
            std::stringstream sql;
            sql << "insert into queue_table values(";
            sql << "'" << queue_ptr->_name << "',";
            sql << queue_ptr->_durable << ",";
            sql << queue_ptr->_exclusive << ",";
            sql << queue_ptr->_auto_delete << ",'";
            sql << queue_ptr->getArgs() << "');";
            return _sql_helper.exec(sql.str(), nullptr, nullptr);
        }

        void remove(const std::string &name)
        {
            // delete from queue_table where name='queue1';
            std::stringstream sql;
            sql << "delete from queue_table where name='";
            sql << name << "';";
            _sql_helper.exec(sql.str(), nullptr, nullptr);
        }

        QueueMap recovery()
        {
            QueueMap result;
            std::string sql = "select name,durable,exclusive,auto_delete,args from queue_table;";
            _sql_helper.exec(sql, selectCallback, &result);
            return result;
        }

        ~MsgQueueMapper()
        {
        }

    private:
        static int selectCallback(void *args, int numcol, char **row, char **fields)
        {
            QueueMap *result = (QueueMap *)args;
            MsgQueue::ptr mqp = std::make_shared<MsgQueue>();
            mqp->_name = row[0];
            mqp->_durable = (bool)std::stoi(row[1]);
            mqp->_exclusive = (bool)std::stoi(row[2]);
            mqp->_auto_delete = (bool)std::stoi(row[3]);
            if (row[4])
                mqp->setArgs(row[4]);
            result->insert({mqp->_name, mqp});
            return 0;
        }

    private:
        SqliteHelper _sql_helper;
    };

    class MsgQueueManager
    {
    public:
        using ptr = std::shared_ptr<MsgQueueManager>;
        MsgQueueManager(const std::string &dbfile)
            : _mapper(dbfile)
        {
            // 启动时恢复历史数据
            _msg_queues = _mapper.recovery();
        }
        bool declareQueue(const std::string &name, bool durable, bool exclusive, bool auto_delete,
                          const google::protobuf::Map<std::string, std::string> &args)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _msg_queues.find(name);
            if (it != _msg_queues.end())
            {
                // 如果队列已经存在，那就直接返回，不需要重复新增
                return true;
            }
            MsgQueue::ptr mqp = std::make_shared<MsgQueue>(name, durable, exclusive, auto_delete, args);
            _msg_queues.insert({name, mqp});
            if (durable == true)
            {
                return _mapper.insert(mqp);
            }
            return true;
        }
        void deleteQueue(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _msg_queues.find(name);
            if (it == _msg_queues.end())
            {
                return;
            }
            if (it->second->_durable == true)
                _mapper.remove(name);
            _msg_queues.erase(name);
        }

        MsgQueue::ptr selectQueue(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _msg_queues.find(name);
            if (it == _msg_queues.end())
            {
                return MsgQueue::ptr();
            }
            return it->second;
        }

        const QueueMap &allQueues()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _msg_queues;
        }

        bool exists(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _msg_queues.find(name);
            if (it == _msg_queues.end())
            {
                // 如果队列不存在，那就直接返回false
                return false;
            }
            return true;
        }

        size_t size()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _msg_queues.size();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _mapper.removeTable();
            _msg_queues.clear();
        }

        ~MsgQueueManager()
        {
        }

    private:
        std::mutex _mtx;
        MsgQueueMapper _mapper;
        QueueMap _msg_queues;
    };
}